package aof

import (
	"abdis/interface/database"
	"abdis/lib/logger"
	"abdis/lib/utils"
	"abdis/redis/connection"
	"abdis/redis/parser"
	"abdis/redis/protocol"
	"context"
	"io"
	"os"
	"strconv"
	"sync"
)

const (
	FsyncAlways = iota
	FsyncEverySec
	FsyncNo
)

type payload struct {
	cmd   database.CmdLine
	index int
	//wg    *sync.WaitGroup
}

type Persister struct {
	ctx       context.Context
	cancel    context.CancelFunc
	db        database.DB
	makeTmpDB func() database.DB

	file *os.File
	path string

	ch       chan *payload
	finished chan struct{}
	buffer   []database.CmdLine

	pausing sync.Mutex

	curDbIndex int
	fsync      int
}

func MakePersister(db database.DB, path string, fsync int, maker func() database.DB) (*Persister, error) {

	ctx, cancel := context.WithCancel(context.Background())

	persister := &Persister{
		path:       path,
		ctx:        ctx,
		cancel:     cancel,
		db:         db,
		makeTmpDB:  maker,
		ch:         make(chan *payload),
		finished:   make(chan struct{}),
		fsync:      fsync,
		curDbIndex: 0,
		buffer:     make([]database.CmdLine, 0),
	}

	persister.Load()

	err := persister.openFile()
	if err != nil {
		return nil, err
	}

	go persister.startListening()

	return persister, nil
}

func (p *Persister) Close() {
	if p.file != nil {
		close(p.ch)
		<-p.finished // block util over
		err := p.file.Close()
		if err != nil {
			logger.Warn(err)
		}
	}
	p.cancel()
}

func (p *Persister) Fsync() {
	p.pausing.Lock()
	defer p.pausing.Unlock()
	if err := p.file.Sync(); err != nil {
		logger.Warn(err)
	}
}

func (p *Persister) Load() {

	aofCh := p.ch
	p.ch = nil
	defer func() { p.ch = aofCh }()

	file, err := os.Open(p.path)
	if err != nil {
		return
	}
	defer file.Close()

	ch := parser.ParseStream(file)
	conn := connection.MakeFakeConn()
	for load := range ch {
		if load.Err != nil {
			if load.Err == io.EOF {
				break
			}
			logger.Error("parse error: " + load.Err.Error())
			continue
		}
		if load.Data == nil {
			continue
		}
		r, ok := load.Data.(*protocol.MultiBulkReply)
		if !ok {
			continue
		}
		ret := p.db.Exec(conn, r.Args)
		if protocol.IsErrorReply(ret) {
			logger.Error("exec err", string(ret.ToBytes()))
		}
	}
}

func (p *Persister) SaveCmd(index int, cmd database.CmdLine) {
	if p.ch == nil {
		return
	}

	load := &payload{
		cmd:   cmd,
		index: index,
	}
	if p.fsync == FsyncAlways {
		p.write(load)
		return
	}

	p.ch <- load
}

func (p *Persister) startListening() {
	for load := range p.ch {
		p.write(load)
	}
	p.finished <- struct{}{}
}

func (p *Persister) write(load *payload) {
	p.pausing.Lock()
	defer p.pausing.Unlock()

	p.buffer = p.buffer[:0]

	if p.curDbIndex != load.index {
		cmd := utils.ToCmdLine("Select", strconv.Itoa(load.index))
		err := p.writeCmd(cmd)
		if err != nil {
			logger.Warn(err)
			return
		}
		// 写下切换数据库了，需要立即改变
		p.curDbIndex = load.index
	}

	err := p.writeCmd(load.cmd)
	if err != nil {
		logger.Warn(err)
		return
	}

	if p.fsync == FsyncAlways {
		err = p.file.Sync()
		if err != nil {
			logger.Warn(err)
		}
	}
}

func (p *Persister) writeCmd(cmd database.CmdLine) error {
	p.buffer = append(p.buffer, cmd)
	data := protocol.MakeMultiBulkReply(cmd).ToBytes()
	_, err := p.file.Write(data)
	return err
}

func (p *Persister) fsyncEverySecond() {

}

func (p *Persister) openFile() error {
	file, err := os.OpenFile(p.path, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0600)
	if err != nil {
		return err
	}
	p.file = file
	return nil
}
